package com.pw.study.flink.chapter9;

import com.pw.study.flink.entities.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

public class MyCep {
    public static void main(String[] args) {
        cep();
    }

    private static void cep() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        //输入数据
        SingleOutputStreamOperator<WaterSensor> data = env.readTextFile("data/file/sensor.txt").map(vc -> {
            String[] line = vc.split(",");
            return new WaterSensor(line[0], Long.valueOf(line[1]), Integer.valueOf(line[2]));
        });
        //水印
        SingleOutputStreamOperator<WaterSensor> stream = data.assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((v, ts) -> v.getTs() * 1000));
        //2.定义规则
        // 简单模式
        Pattern<WaterSensor, WaterSensor> pattern = mc1(stream);
        //匹配次数
        Pattern<WaterSensor, WaterSensor> p2 = mc2(stream);
        //di迭代
        Pattern<WaterSensor, WaterSensor> p3 = mc3(stream);
        //多条件
        Pattern<WaterSensor, WaterSensor> p4 = mc4();

        Pattern<WaterSensor, WaterSensor> p5 = mc5();
        Pattern<WaterSensor, WaterSensor> p6 = mc6();
        Pattern<WaterSensor, WaterSensor> p7 = mc7();
        Pattern<WaterSensor, WaterSensor> p8 = mc8();
        Pattern<WaterSensor, WaterSensor> p9 = mc9();
        Pattern<WaterSensor, WaterSensor> p10 = mc10();
        Pattern<WaterSensor, WaterSensor> p11 = mc11();
        Pattern<WaterSensor, WaterSensor> p12 = mc12();
        Pattern<WaterSensor, WaterSensor> p13 = mc13();
        Pattern<WaterSensor, WaterSensor> p14 = mc14();
        Pattern<WaterSensor, WaterSensor> p15 = mc15();
        Pattern<WaterSensor, WaterSensor> p16 = mc16();
        Pattern<WaterSensor, WaterSensor> p17 = mc17();
        Pattern<WaterSensor, WaterSensor> p = mc18();

        //3。绑定数据,取出数据
       // endMc(stream, p);
        endMc2(stream, p);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void endMc2(SingleOutputStreamOperator<WaterSensor> stream, Pattern<WaterSensor, WaterSensor> p) {
        PatternStream<WaterSensor> ps = CEP.pattern(stream, p);
        OutputTag<WaterSensor> outputTag = new OutputTag<WaterSensor>("timeout") {};
        SingleOutputStreamOperator<String> normal = ps.select(outputTag,
                new PatternTimeoutFunction<WaterSensor, WaterSensor>() {
                    @Override
                    public WaterSensor timeout(Map<String, List<WaterSensor>> pattern, long timeoutTimestamp) throws Exception {
                        System.out.println("timeout...........");
                        return pattern.get("start").get(0);
                    }
                }, (PatternSelectFunction<WaterSensor, String>) map -> map.toString());
        normal.print("匹配成功...");
        normal.getSideOutput(outputTag).print("timeout...");
    }

    private static Pattern<WaterSensor, WaterSensor> mc18() {
        Pattern<WaterSensor, WaterSensor> pattern = Pattern
                .<WaterSensor>begin("start")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value) throws Exception {
                        return "sensor_1".equals(value.getId());
                    }
                })
                .next("end")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value) throws Exception {
                        return "sensor_2".equals(value.getId());
                    }
                })
                .within(Time.seconds(2));
        return pattern;
    }

    private static Pattern<WaterSensor, WaterSensor> mc17() {
        Pattern<WaterSensor, WaterSensor> pattern = Pattern
                .begin(Pattern
                        .<WaterSensor>begin("start")
                        .where(new SimpleCondition<WaterSensor>() {
                            @Override
                            public boolean filter(WaterSensor value) throws Exception {
                                return "sensor_1".equals(value.getId());
                            }
                        })
                        .next("next")
                        .where(new SimpleCondition<WaterSensor>() {
                            @Override
                            public boolean filter(WaterSensor value) throws Exception {
                                return "sensor_2".equals(value.getId());
                            }
                        }))
                .times(2);
return pattern;
    }

    private static Pattern<WaterSensor, WaterSensor> mc16() {
        Pattern<WaterSensor, WaterSensor> pattern = Pattern
                .<WaterSensor>begin("start")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value) throws Exception {
                        return "sensor_1".equals(value.getId());
                    }
                }).times(2).optional()  // 0次或2次
                .next("end")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value) throws Exception {
                        return "sensor_2".equals(value.getId());
                    }
                });
        return pattern;
    }

    private static Pattern<WaterSensor, WaterSensor> mc15() {
        // 2. 指定规则(定义模式)
        Pattern<WaterSensor, WaterSensor> pattern = Pattern
                .<WaterSensor>begin("s1")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value) throws Exception {
                        return "sensor_1".equals(value.getId());
                    }
                })
                .times(2)
                //.consecutive()
                //.optional() // 模式的可选性
                .greedy()
                .next("s2")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor value) throws Exception {
                        return "sensor_2".equals(value.getId());
                    }
                });
        return pattern;
    }

    private static Pattern<WaterSensor, WaterSensor> mc14() {
        Pattern<WaterSensor, WaterSensor> p = Pattern.<WaterSensor>begin("s1").where(new SimpleCondition<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor sensor) throws Exception {
                return sensor.getId().equals("sensor_1");
            }
        }).times(2).allowCombinations();
        return p;
    }

    private static Pattern<WaterSensor, WaterSensor> mc13() {
        Pattern<WaterSensor, WaterSensor> p = Pattern.<WaterSensor>begin("s1").where(new SimpleCondition<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor sensor) throws Exception {
                return sensor.getId().equals("sensor_1");
            }
        }).times(2).consecutive();
        return p;
    }

    private static Pattern<WaterSensor, WaterSensor> mc12() {
        Pattern<WaterSensor, WaterSensor> p = Pattern.<WaterSensor>begin("start").where(new SimpleCondition<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor sensor) throws Exception {
                return sensor.getId().equals("sensor_1");
            }
        }).notNext("s2").where(new SimpleCondition<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor sensor) throws Exception {
                return sensor.getId().equals("sensor_2");
            }
        }).followedBy("end").where(new SimpleCondition<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor sensor) throws Exception {
                return sensor.getId().equals("sensor_3");
            }
        });
        return p;
    }

    private static Pattern<WaterSensor, WaterSensor> mc11() {
        Pattern<WaterSensor, WaterSensor> p = Pattern.<WaterSensor>begin("start").where(new SimpleCondition<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor sensor) throws Exception {
                return sensor.getId().equals("sensor_1");
            }
        }).followedBy("end").where(new SimpleCondition<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor sensor) throws Exception {
                return sensor.getId().equals("sensor_2");
            }
        });
        return p;
    }

    private static Pattern<WaterSensor, WaterSensor> mc10() {
        Pattern<WaterSensor, WaterSensor> p = Pattern.<WaterSensor>begin("s1").where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getId().equals("sensor_1");
                    }
                }).notNext("next")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getId().equals("sensor_2");
                    }
                });
        return p;
    }

    private static Pattern<WaterSensor, WaterSensor> mc9() {
        Pattern<WaterSensor, WaterSensor> p = Pattern.<WaterSensor>begin("s1").where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getId().equals("sensor_1");
                    }
                }).followedBy("end")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getId().equals("sensor_2");
                    }
                });
        return p;
    }

    private static Pattern<WaterSensor, WaterSensor> mc8() {
        Pattern<WaterSensor, WaterSensor> p = Pattern.<WaterSensor>begin("s1").where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getId().equals("sensor_1");
                    }
                }).next("end")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getId().equals("sensor_2");
                    }
                });
        return p;
    }

    private static Pattern<WaterSensor, WaterSensor> mc7() {
        Pattern<WaterSensor, WaterSensor> p = Pattern.<WaterSensor>begin("s1").where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getId().equals("sensor_1");
                    }
                }).timesOrMore(2)
                .until(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getVc() > 55;
                    }
                });
        return p;
    }

    private static Pattern<WaterSensor, WaterSensor> mc6() {
        Pattern<WaterSensor, WaterSensor> p = Pattern.<WaterSensor>begin("s1").where(new SimpleCondition<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor sensor) throws Exception {
                return sensor.getId().equals("sensor_1");
            }
        }).timesOrMore(2).until(new SimpleCondition<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor sensor) throws Exception {
                return sensor.getVc() > 40;
            }
        });
        return p;
    }

    private static Pattern<WaterSensor, WaterSensor> mc5() {
        Pattern<WaterSensor, WaterSensor> p = Pattern.<WaterSensor>begin("s1")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getId().equals("sensor_1");
                    }
                }).timesOrMore(2);
        return p;
    }

    private static Pattern<WaterSensor, WaterSensor> mc4() {
        Pattern<WaterSensor, WaterSensor> p = Pattern.<WaterSensor>begin("s1")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getId().equals("sensor_1");
                    }
                })
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getVc() > 55;
                    }
                })
                .or(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor sensor) throws Exception {
                        return sensor.getTs() > 4;
                    }
                });
        return p;
    }

    private static Pattern<WaterSensor, WaterSensor> mc3(SingleOutputStreamOperator<WaterSensor> stream) {
        Pattern<WaterSensor, WaterSensor> s1 = Pattern.<WaterSensor>begin("s1").where(new IterativeCondition<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor value, Context<WaterSensor> ctx) throws Exception {
                System.out.println("ctx: " + ctx);
                return value.getId().equals("sensor_1");
            }
        });
        return s1;
    }

    private static Pattern<WaterSensor, WaterSensor> mc2(SingleOutputStreamOperator<WaterSensor> stream) {
        Pattern<WaterSensor, WaterSensor> s1 = Pattern.<WaterSensor>begin("s1").where(new SimpleCondition<WaterSensor>() {

            @Override
            public boolean filter(WaterSensor sensor) throws Exception {
                return sensor.getVc() > 50;
            }
        }).oneOrMore();
        //.timesOrMore(5);
        //.times(2);
        //.times(2, 5);
        return s1;
    }

    private static Pattern<WaterSensor, WaterSensor> mc1(SingleOutputStreamOperator<WaterSensor> stream) {
        //2.定义规则
        Pattern<WaterSensor, WaterSensor> pattern = Pattern.<WaterSensor>begin("start").where(new SimpleCondition<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor sensor) throws Exception {
                System.out.println("==============================");
                return sensor.getId().equals("sensor_1");
            }
        });
        return pattern;
    }

    private static void endMc(SingleOutputStreamOperator<WaterSensor> stream, Pattern<WaterSensor, WaterSensor> pattern) {
        //3.绑定数据流
        PatternStream<WaterSensor> ps = CEP.pattern(stream, pattern);
        //4.取出数据
        ps.select(map -> map.toString()).print();
    }
}
